package t20240424;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.redis.command.RedisCommand;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import static org.apache.flink.streaming.connectors.redis.config.RedisValidator.REDIS_COMMAND;

public class TestRedis {


    private static final String REDIS_HOST = "127.0.0.1";
    private static final String REDIS_PORT = "6379";
    private static final String REDIS_PASSWORD = "yy2024";

    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        String ddl =
                "create table sink_redis(username varchar, level varchar, age varchar, fieldkey varchar, fieldvalue varchar) with ( 'connector'='redis', "
                        + "'host'='"
                        + REDIS_HOST
                        + "','port'='"
                        + REDIS_PORT
                        + "', 'redis-mode'='single','password'='"
                        + REDIS_PASSWORD
                        + "','"
                        + REDIS_COMMAND
                        + "'='"
                        + RedisCommand.HMSET
                        + "',  'minIdle'='1'  )";

        tableEnv.executeSql(ddl);
        String sql = " insert into sink_redis select * from (values ('test_hash', '3', '18', 'city', '广州'))";
        tableEnv.executeSql(sql);

    }
}
